package com.xdja.kafka.file.sink;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
import java.util.Collection;
import java.util.Map;

/**
 * 从kafka的topic中读取文件内容，并写入到文件的任务类
 */
public class FileSinkTask extends SinkTask {
    // 声明一个日志对象

    // 声明一个文件名变量
    private String filename;
    // 声明一个输出流对象
    private PrintStream outputStream;

    /**
     * 构造函数
     */
    public FileSinkTask() {
        System.out.println("FileSinkTask构造方法1");
    }

    /**
     * 构造函数
     * @param outputStream
     */
    public FileSinkTask(PrintStream outputStream) {
        System.out.println("FileSinkTask构造方法2");
        filename = null;
        this.outputStream = outputStream;
    }

    /**
     * 获取版本号
     * @return
     */
    public String version() {
        String version = new FileSinkConnector().version();
        System.out.println("调用了FileSinkTask()的version() " + version);
        return version;
    }

    /**
     * tast启动时夹子啊的方法
     * @param props
     */
    public void start(Map<String, String> props) {
        System.out.println("调用了FileSinkTask的start() " + props);
        filename = props.get(FileSinkConnector.FILE_CONFIG);
        if (filename == null) {
            outputStream = System.out;
        } else {
            try {
                outputStream = new PrintStream(new FileOutputStream(filename, true), false, "UTF-8");
            } catch (FileNotFoundException e) {
                throw new ConnectException("Couldn't find or create file for FileStreamSinkTask", e);
            } catch (UnsupportedEncodingException e) {
                throw new ConnectException("Couldn't find or create file for FileStreamSinkTask", e);
            }
        }
    }

    /**
     * 将topic内容写入文件输出流
     * @param sinkRecords
     */
    public void put(Collection<SinkRecord> sinkRecords) {
        System.out.println("调用了FileSinkTask的put() " );
        for (SinkRecord record : sinkRecords) {
            outputStream.println(record.value());
        }
    }

    /**
     * 将读取到的记录，提交给文件
     * @param offsets
     */
    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        System.out.println("调用了FileSinkTask的flush() " );
        outputStream.flush();
    }

    /**
     * task停止时加载的方法
     */
    @Override
    public void stop() {
        System.out.println("调用了FileSinkTask的stop() " );
        if (outputStream != null && outputStream != System.out)
            outputStream.close();
    }


}
